package com.zwy.rxjava;

import android.app.Activity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.view.View.OnClickListener;
import android.widget.Button;
import android.widget.TextView;

import rx.Observable;
import rx.Subscriber;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

/**
 * 多线程
 */
public class RxJavaDemo4 extends Activity implements OnClickListener {
	private TextView tvs;
	private TextView tve;
	private TextView tvc;
	private Button btn1;
	private Button btn2;
	private Button btn3;
	private int i ;

	@Override
	protected void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		
		setContentView(R.layout.activity_rxjava2);
		
		tvs = (TextView) findViewById(R.id.tvs_rx2);
		tve = (TextView) findViewById(R.id.tve_rx2);
		tvc = (TextView) findViewById(R.id.tvc_rx2);
		
		btn1 = (Button) findViewById(R.id.btn1_rx2);
		btn2 = (Button) findViewById(R.id.btn2_rx2);
		btn3 = (Button) findViewById(R.id.btn3_rx2);
		
		btn1.setOnClickListener(this);
		btn2.setOnClickListener(this);
		btn3.setOnClickListener(this);
		
		findViewById(R.id.btn4_rx2).setOnClickListener(this);

	}

	@Override
	public void onClick(View v) {
		Log.i("zwy", "0t:" + Thread.currentThread().getId());
		switch (v.getId()) {
		case R.id.btn1_rx2:
			test1();
			break;
		case R.id.btn2_rx2:
			test2();
			break;
		case R.id.btn3_rx2:
			test3();
			break;
		case R.id.btn4_rx2:
			test4();
			break;
		default:
			break;
		}
	}

	/**
	 * 使用Subscriber来执行回调
	 * subscribeOn(Schedulers.io())指定Observable执行在子线程
	 * observeOn(AndroidSchedulers.mainThread())指定subscribe执行在Android主线程
	 */
	private void test1() {
	
		Observable.create(new Observable.OnSubscribe<String>() {
			@Override
			public void call(Subscriber<? super String> subscriber) {
				Log.i("zwy", "1t:" + Thread.currentThread().getId());
				if (i%3==0) {
					subscriber.onNext("MultiThreading");
				}else if((i%3==1)){
					subscriber.onCompleted();
				}else{
					subscriber.onError(null);
				}
				i++;
			}
		}).subscribeOn(Schedulers.io())///
		.observeOn(AndroidSchedulers.mainThread())//
		.subscribe(new Subscriber<String>() {//

			@Override
			public void onCompleted() {
				Log.i("zwy", "4t:" + Thread.currentThread().getId());
				tvs.setText("完成！！！！");
			}

			@Override
			public void onError(Throwable arg0) {
				Log.i("zwy", "3t:" + Thread.currentThread().getId());
				tvs.setText("失败！！！" );
			}

			@Override
			public void onNext(String arg0) {
				Log.i("zwy", "2t:" + Thread.currentThread().getId());
				tvs.setText(arg0);
			}
		});
	}
	/**
	 * 使用Action1来执行回调
	 */
	private void test2() {
		Observable.create(new Observable.OnSubscribe<String>() {
			@Override
			public void call(Subscriber<? super String> subscriber) {
				Log.i("zwy", "1t:" + Thread.currentThread().getId());
				if (i%3==0) {
					subscriber.onNext("MultiThreading");
				}else if((i%3==1)){
					subscriber.onCompleted();
				}else{
					subscriber.onError(null);
				}
				i++;
			}
		}).subscribeOn(Schedulers.io())//
				.observeOn(AndroidSchedulers.mainThread())//
				.subscribe(new Action1<String>() {//
			@Override
			public void call(String arg0) {
				Log.i("zwy", "2t:" + Thread.currentThread().getId());
				tvs.setText(arg0);
			}
		}, new Action1<Throwable>() {

			@Override
			public void call(Throwable arg0) {
				Log.i("zwy", "3t:" + Thread.currentThread().getId());
				tvs.setText("失败！！！");
			}
		}, new Action0() {

			@Override
			public void call() {
				Log.i("zwy", "4t:" + Thread.currentThread().getId());
				tvs.setText("完成！！！！");
			}
		});
	}
	/**
	 * 使用操作符map，操作符中的线程跟Observable是一样的,在这里都是异步
	 */
	private void test3(){
		Observable.create(new Observable.OnSubscribe<MessageEvent>() {
			@Override
			public void call(Subscriber<? super MessageEvent> subscriber) {
				Log.i("zwy", "1t:" + Thread.currentThread().getId());
				if (i%3==0) {
					subscriber.onNext(new MessageEvent("Hello RxAndroid!!!!!"));
				}else if((i%3==1)){
					subscriber.onCompleted();
				}else{
					subscriber.onError(null);
				}
				i++;
			}
		}).map(new Func1<MessageEvent, MessageEvent>() {

			@Override
			public MessageEvent call(MessageEvent arg0) {
				Log.i("zwy", "5t:" + Thread.currentThread().getId());
				return new MessageEvent(arg0.getMsg()+"Hello Winson!!!");
			}
		}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<MessageEvent>() {

			@Override
			public void onCompleted() {
				Log.i("zwy", "4t:" + Thread.currentThread().getId());
				tvs.setText("完成！！！！");
			}

			@Override
			public void onError(Throwable arg0) {
				Log.i("zwy", "3t:" + Thread.currentThread().getId());
				tvs.setText("失败！！！" );
			}

			@Override
			public void onNext(MessageEvent arg0) {
				Log.i("zwy", "2t:" + Thread.currentThread().getId());
				tvs.setText(arg0.getMsg());
			}
		});
	}
	/**
	 * onNext回调拓展
	 */
	private void test4(){
		
		Observable.create(new Observable.OnSubscribe<MessageTask>() {
			@Override
			public void call(Subscriber<? super MessageTask> subscriber) {
				Log.i("zwy", "1t:" + Thread.currentThread().getId());
				
				try {
					subscriber.onNext(new MessageTask(0, "开始执行任务。。。。。"));
					Thread.sleep(1000);
					subscriber.onNext(new MessageTask(1, "正在执行任务。。。。"));
					Thread.sleep(1000);
					subscriber.onNext(new MessageTask(2, "还在执行任务。。。。"));
					Thread.sleep(1000);
					subscriber.onCompleted();
					
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				
			}
		}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Subscriber<MessageTask>() {

			@Override
			public void onCompleted() {
				Log.i("zwy", "4t:" + Thread.currentThread().getId());
				tvs.setText("已经完成了！！！！");
			}

			@Override
			public void onError(Throwable arg0) {
				Log.i("zwy", "3t:" + Thread.currentThread().getId());
				tvs.setText("可惜失败了！！！" );
			}

			@Override
			public void onNext(MessageTask arg0) {
				Log.i("zwy", "2t:" + Thread.currentThread().getId());
				switch (arg0.getTag()) {
				case 0:
					tvs.setText(arg0.getMsg()+"我要准备");
					break;
				case 1:
					tvs.setText(arg0.getMsg()+"已经开始了呀");
					break;
				case 2:
					tvs.setText(arg0.getMsg()+"还要多久呀！");
					break;

				default:
					break;
				}
			}
		});
	}
}
